package com.yupi.springbootinit.sse;

import com.yupi.springbootinit.common.ErrorCode;
import com.yupi.springbootinit.exception.BusinessException;
import com.yupi.springbootinit.model.entity.Chart;
import lombok.extern.slf4j.Slf4j;
import org.jetbrains.annotations.NotNull;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.server.ServerHttpResponse;
import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;

import static com.yupi.springbootinit.constant.CommonConstant.CHART_SSE_ID;

@Service
@Slf4j
public class SseService {

    /**
     * sse连接池
     */
    private final Map<Object, SseEmitter> sseEmitterPool = new ConcurrentHashMap<>();


    public Map<Object, SseEmitter> getSseEmitterPool() {
        return sseEmitterPool;
    }


    /**
     * 根据id获取sse连接，当sse连接不存在时，创建
     * @param id
     * @return
     */
    public SseEmitter getConn(String id) {
        SseEmitter sseEmitter = sseEmitterPool.get(id);
        if(sseEmitter==null){
            sseEmitter = new SseEmitter(-1L){
                @Override
                protected void extendResponse(@NotNull ServerHttpResponse outputMessage) {
                    super.extendResponse(outputMessage);
                    //重写-返回编码改为utf8
                    HttpHeaders headers = outputMessage.getHeaders();
                    headers.setContentType(new MediaType(MediaType.TEXT_EVENT_STREAM, StandardCharsets.UTF_8));
                }
            };//永不超时
            sseEmitter.onCompletion(() -> {
                log.info("sse完成");
                sseEmitterPool.remove(id);
            });
            sseEmitter.onTimeout(() -> {
                log.warn("sse超时");
                sseEmitterPool.remove(id);
            });
            sseEmitterPool.put(id, sseEmitter);
        }
        return sseEmitter;
    }

    /**
     * 发送消息
     * @param id sse连接id
     * @param content
     */
    public void sendMessage(Object id, Object content) {
        SseEmitter sseEmitter = sseEmitterPool.get(id);
        if (sseEmitter == null) {
            log.warn("sendMessage：还未连接到sse");
//            throw new BusinessException(ErrorCode.SYSTEM_ERROR, "还未连接到sse");
            return;
        }
        //执行耗时操作
        CompletableFuture.runAsync(() -> {
            try {
                sseEmitter.send(content, MediaType.APPLICATION_JSON);
            } catch (Exception e) {
                throw new BusinessException(ErrorCode.SYSTEM_ERROR,"sse发送消息失败");
            }
        });
    }



    /**
     * sse发送消息
     * @param chart
     */
    public void sendChartStatusMessage(Chart chart) {
        sendMessage(CHART_SSE_ID,chart);

    }
}